package com.itheima.kafka.boot.listen;

import com.itheima.kafka.boot.stream.KafkaStreamListener;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.springframework.stereotype.Component;

import java.util.Arrays;

/**
 * @Description:
 * @Version: V1.0
 */
@Component
public class StreamListener implements KafkaStreamListener<KStream<String,String>> {

    @Override
    public String listenerTopic() {
        return "itcast-topic-input";
    }

    @Override
    public String sendTopic() {
        return "itcast-topic-output";
    }

    @Override
    public KStream<String,String> getService(KStream<String,String> stream) {
        return stream.flatMapValues( value -> {
            // value 接收消息的具体内容
            return Arrays.asList(value.split(" "));
        })
         // 根据value进行分组
        .groupBy((key,value)->value)
        // 聚合计算时间间隔
        .windowedBy(TimeWindows.of(5000))
        // 聚合查询：求单词总个数
        .count()
        // 转成 KStream
        .toStream()
        // key和value转成string
        .map((key, value) -> {
            return new KeyValue<>(key.key().toString(), value.toString());
        });
    }
}
